package com.wyz.socket.core.impl.stealing;

import com.wyz.socket.core.base.IoTask;

import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.IntFunction;
import java.util.stream.Stream;

/**
 * 窃取调度服务
 * 窃取模式的详细
 * <p>
 * 1 2 3 三个线程
 * 如果1 3 线程空闲
 * 2 线程正在繁忙
 * 这样通过 窃取分担2 线程的任务
 */
public class StealingService implements Closeable {

    private final int minSafetyThreshold;

    private final IoStealingSelectorThread[] threads;

    private final Queue<IoTask>[] queues;

    private volatile boolean isCLose;

    public StealingService(IoStealingSelectorThread[] threads, int minSafetyThreshold) {
        this.threads = threads;
        this.minSafetyThreshold = minSafetyThreshold;
        Stream<IoStealingSelectorThread> stream = Arrays.stream(threads);
        Stream<Queue<IoTask>> linkedBlockingQueueStream = stream.map(IoStealingSelectorThread::getReadyTaskQueue);
        this.queues = linkedBlockingQueueStream.toArray((IntFunction<Queue<IoTask>[]>) ArrayBlockingQueue[]::new);
    }

    IoTask steal(final Queue<IoTask> excludedQueue) {
        final int minSafetyThreshold = this.minSafetyThreshold;
        final Queue<IoTask>[] queues = this.queues;

        for (Queue<IoTask> queue : queues) {
            if (queue == excludedQueue) {
                continue;
            }
            int size = queue.size();
            if (size > minSafetyThreshold) {
                IoTask poll = queue.poll();
                if (poll != null) {
                    return poll;
                }
            }
        }
        return null;
    }

    public IoStealingSelectorThread getNotBusyThread() {
        IoStealingSelectorThread targetThread = null;
        final IoStealingSelectorThread[] threads = this.threads;
        long targetSize = Long.MAX_VALUE;
        for (IoStealingSelectorThread item : threads) {
            long registerKeyCount = item.getBusyStatus();
            if (registerKeyCount != -1 && registerKeyCount < targetSize) {
                targetSize = registerKeyCount;
                targetThread = item;
            }
        }
        return targetThread;
    }

    public void execute(IoTask task) {
    }

    public boolean isCLose() {
        return isCLose;
    }

    @Override
    public void close() throws IOException {
        if (isCLose) {
            return;
        }
        isCLose = true;
        for (IoStealingSelectorThread item : threads) {
            item.close();
        }
    }
}
